package com

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/13 23:32
  *
  *
  *                  单词计数流处理
  */
object StreamWordCount {
  def main(args: Array[String]): Unit = {

    //创建一个批处理的执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    //从外部命令中提取参数，作为socket的主机名和参数
    val paramTool: ParameterTool = ParameterTool.fromArgs(args)
    val host: String = paramTool.get("host")
    val port: Int = paramTool.getInt("port")




    //从文件中读取数据
    val inputDataStream: DataStream[String] = env.socketTextStream(host, port)


    val resultDataStream: DataStream[(String, Int)] = inputDataStream
      .flatMap(_.split(" "))
      .filter(_.nonEmpty).disableChaining() //拆开Task
      .map((_, 1)).startNewChain()  //开始Task
      .keyBy(0)
      .sum(1)
    resultDataStream.print().setParallelism(1)


    //启动任务执行
    env.execute("wordcount stream");


  }

}
